from pythondemo.Pandas_Tushare import get_tushare_stock as gts
import threading
from pythondemo.Pandas_Tushare import config_mysql as cm
import math
import datetime
import time
import pandas as pd
from multiprocessing import Process


# 重写多线程，便于获得返回值
class MoreThread(threading.Thread):

    def __init__(self, func, args='', name=''):
        threading.Thread.__init__(self)
        self.func = func
        self.args = args
        self.name = name
        self.result = self.func(*self.args)

    def get_result(self):
        try:
            return self.result
        except Exception as e:
            return e


class MoreProcess(Process):

    def __init__(self, func, args='', name=''):
        Process.__init__(self)
        self.func = func
        self.args = args
        self.name = name
        self.result = self.func(*self.args)

    def get_result(self):
        try:
            return self.result
        except Exception as e:
            return e


def channel_calc(start, stock_data, code):
    '''
    首先获得20日数据保持
    在20日的基础上加10日，获得数据保持
    比较前后大小，
    如果后者比前者大 ，用后者的数据替换前者的，30日替换20日的数据
    计数器加10，获取数据，与前面的30日数据比较，以此类推
    '''
    vessel_stock = pd.DataFrame()

    count = datetime.timedelta(days=10)
    count_twenty = datetime.timedelta(days=20)
    for cou in range(10):

        # 查找阶段最低价作为参照
        # for code_low in ts_codes:
        end = start - count_twenty

        rds_low = stock_data[start: end]
        # 统计n条记录中，low列的最小值
        low = float(rds_low['low'].min())
        end_t = end - count
        flags = True
        while flags:

            rds_low = stock_data[start: end_t]
            low_s = float(rds_low["low"].min())
            if low != low_s:
                for l in range(1, len(stock_data), 5):
                    low = low_s
                    end_t = end_t - count

                    rds_low = stock_data[start: end_t]
                    low_s = float(rds_low["low"].min())
                    if low == low_s:
                        break
                    elif len(stock_data) - l < 5:
                        cou = 10
                        break
            low_row = rds_low.loc[rds_low.low == low]
            # 判断是否多日处于低点
            if len(low_row) > 1:
                low_row = low_row[-1:]
            vessel_stock = pd.concat([vessel_stock, low_row])
            # self.engine.put_sql(low_row, code + '_low_data', 'append')
            start = low_row.reset_index().loc[0, 'trade_date']
            flags = False

        # 查找阶段最高价
        end = start - count_twenty
        rds_high = stock_data[start: end_t]
        # 统计n条记录中，high列的最大值
        high = float(rds_high['high'].max())
        end_t = end - count
        flags = True
        while flags:
            # 查找第二组数据，用以和第一组数据对比

            rds_high = stock_data[start: end_t]
            high_s = float(rds_high["high"].max())
            # 如果第二组数据比第一组大，用第二组数据替换第一组数据，继续查找第三组数据用以和第二组数据对比
            if high != high_s or (high - low) / high < 0.1:
                for l in range(1, len(stock_data), 5):
                    high = high_s
                    end_t = end_t - count

                    rds_high = stock_data[start: end_t]
                    high_s = float(rds_high["high"].max())
                    if high == high_s and (high - low) / high < 0.1:
                        break
                    elif len(stock_data) - l < 5:
                        break
            high_row = rds_high.loc[rds_high['high'] == high]
            # 判断是否多日处于高点
            if len(high_row) > 1:
                high_row = high_row[-1:]
            vessel_stock = pd.concat([vessel_stock, high_row])
            # self.engine.put_sql(high_row, code + '_high_data', 'append')
            start = high_row.reset_index().loc[0, 'trade_date']
            flags = False
    return vessel_stock


def channel_calc_from_date(start, stock_data, code):
    vessel_stock = pd.DataFrame()

    count = datetime.timedelta(days=10)
    count_twenty = datetime.timedelta(days=20)
    for cou in range(10):

        # 查找阶段最高价
        end = start - count_twenty

        rds_high = stock_data[start: end]
        # 统计n条记录中，high列的最大值
        high = float(rds_high['high'].max())
        end_t = end - count
        break_flag = True
        while break_flag:
            # 查找第二组数据，用以和第一组数据对比

            rds_high = stock_data[start: end_t]
            high_s = float(rds_high["high"].max())
            # 如果第二组数据比第一组大，用第二组数据替换第一组数据，继续查找第三组数据用以和第二组数据对比
            if high != high_s:
                for l in range(1, len(stock_data), 5):
                    high = high_s
                    end_t = end_t - count

                    rds_high = stock_data[start: end_t]
                    high_s = float(rds_high["high"].max())

                    if high == high_s:
                        break
                    elif len(stock_data) - l < 5:
                        break

            high_row = rds_high.loc[rds_high.high == high]
            # 判断是否多日处于高点
            if len(high_row) > 1:
                high_row = high_row[-1:]
            # print('this is hr:', high_row)
            vessel_stock = pd.concat([vessel_stock, high_row])
            # print('this is vs:', vessel_stock)
            # self.engine.put_sql(high_row, code + '_high_data', 'append')
            start = high_row.reset_index().loc[0, 'trade_date']
            break_flag = False

        # 查找阶段最低价
        end = start - count_twenty

        rds_low = stock_data[start: end]
        # 统计n条记录中，low列的最小值
        low = float(rds_low['low'].min())
        end_t = end - count
        break_flag = True
        while break_flag:

            rds_low = stock_data[start: end_t]
            low_s = float(rds_low["low"].min())
            if low != low_s or (high - low) / high < 0.1:
                for l in range(1, len(stock_data), 5):
                    low = low_s
                    end_t = end_t - count

                    rds_low = stock_data[start: end_t]
                    low_s = float(rds_low["low"].min())
                    if low == low_s and (high - low) / high >= 0.1:
                        break
                    elif len(stock_data) - l < 5:
                        break
            low_row = rds_low.loc[rds_low['low'] == low]
            # 判断是否多日处于低点
            if len(low_row) > 1:
                low_row = low_row[-1:]
            vessel_stock = pd.concat([vessel_stock, low_row])
            # self.engine.put_sql(low_row, code + '_low_data', 'append')
            start = low_row.reset_index().loc[0, 'trade_date']
            break_flag = False

    return vessel_stock


def entry_to_channel(stocks, count, plus):
    sem = threading.Semaphore(28)
    with sem:
        stock_high_low = pd.DataFrame()
        st_time = time.time()
        for i in range(count, plus):

            ts_code = stocks.loc[i, 'ts_code']
            stock_code = ts_code.replace('.S', '_S')
            start_time = datetime.datetime.now().date() - datetime.timedelta(days=1)
            end_time = start_time - datetime.timedelta(days=730)
            now_date = start_time.strftime('%Y%m%d')
            old_date = end_time.strftime('%Y%m%d')
            stock_data = gts.transfer_stock_date(ts_code, old_date, now_date)
            stock_data['trade_date'] = pd.to_datetime(stock_data['trade_date'])
            stock_data.set_index('trade_date', inplace=True)
            end = start_time - datetime.timedelta(days=30)
            rds = stock_data[start_time: end]
            high = float(rds['high'].max())
            high_row = rds.loc[rds['high'] == high]
            if len(high_row) > 1:
                high_row = high_row[-1:]
            high_date = high_row.reset_index().loc[0, 'trade_date']
            low = float(rds['low'].min())
            low_row = rds.loc[rds['low'] == low]
            # 判断是否多日处于低点
            if len(low_row) > 1:
                low_row = low_row[-1:]
            low_row.reset_index(inplace=True)
            low_date = low_row.loc[0, 'trade_date']
            if high_date > low_date:
                flag_ccfd = channel_calc_from_date(start_time, stock_data, stock_code)
                stock_high_low = pd.concat([stock_high_low, flag_ccfd])
            else:
                flag_cc = channel_calc(start_time, stock_data, stock_code)
                stock_high_low = pd.concat([stock_high_low, flag_cc])
        en_time = time.time()
        print('In ', en_time - st_time, ' seconds')
        return stock_high_low


class SelectStock(object):

    def __init__(self):
        self.engine = cm.ConfSQL()

    # 筛选股票
    def select_stock_thread(self):

        def channel_thread():
            codes = self.engine.get_sql('stockCodes')
            thread_len = math.ceil(len(codes) / 142)
            threads = []
            count = 0
            plus = 0
            stock_values = pd.DataFrame()
            lock = threading.Lock()
            for i in range(thread_len):
                # 任务拆分
                tasks = codes[count:count + 142]
                # 创建线程
                plus += 142
                t = MoreThread(func=entry_to_channel, args=(tasks, count, plus))
                threads.append(t)
                # t.start()
                count += 142
                print('当前线程数：', len(threads))
                # 开始线程
            for i in range(len(threads)):
                threads[i].start()
            for i in range(len(threads)):
                threads[i].join()
            for value in range(len(threads)):
                values = threads[value].get_result()
                stock_values = pd.concat([stock_values, values])
            self.engine.put_sql(stock_values, 'stock_channel_high_and_low', 'replace')

        channel_thread()

    # 获取最近20个交易日数据,code样式 600010_SH


class Main(object):

    def __init__(self):
        self.engine = cm.ConfSQL()

    def more_thread(self):
        date_one = getattr(self.engine.read_sql('stock_channel', 1), 'trade_date')
        # 获取最后交易日期
        date_two = gts.GetStock().cal_date()
        # 对比最新交易日与数据库中最后一个交易日
        if date_one != date_two:
            _engine = cm.ConfSQL()
            _stockCode = _engine.get_sql('stockCodes')
            ts_codes = getattr(_stockCode, 'ts_code')
            ts_codes = ts_codes.replace(r'.S', r'_S', regex=True)
            thread_len = math.ceil(len(ts_codes) / 142)
            # ln = len(ts_codes)
            # thread_len = math.ceil(ln / 1988)
            # return_dict = Manager().dict()
            # 线程池
            threads = []
            # 定位坐标器
            count = 0
            # df = pd.DataFrame()
            for i in range(thread_len):
                # 任务拆分
                # tasks = ts_codes[count:count + int(3976 / 2)]
                tasks = ts_codes[count:count + 142]
                # 创建线程
                # t = Process(target=SelectStock().channel_calc, args=(tasks,))
                t = MoreThread(func=channel_calc, args=(tasks,))
                threads.append(t)
                print('当前线程数：', len(threads))
                # count += 142
                count += int(3976 / 2)
            # 开始线程
            for i in range(len(threads)):
                threads[i].start()
            for i in range(len(threads)):
                threads[i].join()
            # for i in range(len(threads)):
            #     flag = threads[i].get_result()
            # df = pd.concat([df, flag])
            # self.engine.publicSQL(df, 'channel', 'replace')
            print('今日数据已更新')
        else:
            print('今日数据已存在')

    def read_data_file(self):
        pass

    def put_stock_data(self):
        codes = self.engine.get_sql('stockCodes')
        thread_len = math.ceil(len(codes) / 142)
        threads = []
        count = 0
        for i in range(thread_len):
            tasks = codes[count:count + 142]

            t = MoreThread(func=entry_to_channel, args=(tasks,))
            threads.append(t)
            print('当前线程数：', len(threads))
            count += 142
            # 开始线程
        for i in range(len(threads)):
            threads[i].start()
        for i in range(len(threads)):
            threads[i].join()


if __name__ == '__main__':
    SelectStock().select_stock_thread()
